package com.rongdong.parkingcapture.client;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.rongdong.parkingcapture.common.R;
import com.rongdong.parkingcapture.domain.Task;
import com.rongdong.parkingcapture.domain.vo.TaskIdVO;
import com.rongdong.parkingcapture.service.impl.TaskServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.RequestBody;

import javax.annotation.Resource;
import java.util.Date;

import static com.rongdong.parkingcapture.constant.Constants.SUCCESS_MESSAGE;
import static com.rongdong.parkingcapture.constant.Constants.TASK_ID;

/**
 * kafka消费者
 */
@Slf4j
@Service
public class KafkaConsumer {

    @Resource
    TaskServiceImpl taskService;

    /**
     * 消费，将task输出进行展示
     */
    //@KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.consumer.group-id1}")
    public void consume(@RequestBody Task task) {
        System.out.println("Received : " + task);
    }

    /**
     * 将监听到的消息，进行入库
     */
    @KafkaListener(topics = "${spring.kafka.topic}", groupId = "${spring.kafka.consumer.group-id2}")
    public R<TaskIdVO> store(@RequestBody Task task) {
        try {
            task.setCreateTime(new Date());
            //查询数据库是否有该taskId
            QueryWrapper<Task> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq(TASK_ID, task.getTaskId());
            boolean exists = taskService.exists(queryWrapper);
            if (exists) {
                return R.fail("add task failure");
            }

            //存入数据，将json对象类型转换为字符串类型
            boolean isSave = taskService.save(taskService.objectToString(task));
            if (!isSave) {
                return R.fail("add task failure");
            }
            //返回值处理
            TaskIdVO taskIdVO = new TaskIdVO();
            taskIdVO.setTaskId(task.getTaskId());
            return R.ok(taskIdVO, SUCCESS_MESSAGE);
        } catch (DataIntegrityViolationException e) {
            return R.fail("Data integrity violation: " + e.getMessage());
        }
    }

}
